#include "ThreadPool.h"

using namespace std;
using namespace util;

ThreadPool::ThreadPool(int taskMax, int threadMin, int threadMax)
{
    if (taskMax <= 0)
        taskMax = 1024;
    cfg_.taskMax_ = taskMax;

    if (threadMin <= 0)
        threadMin = 2;
    cfg_.threadMin_ = threadMin;

    if (threadMax <= 0)
        threadMax = std::thread::hardware_concurrency();
    cfg_.threadMax_ = threadMax;

    for (int i = 0; i < threadMin; i ++ )
    {
        auto threadptr = std::make_unique<Thread>(std::bind(&ThreadPool::consumerFunc, this, std::placeholders::_1));
        const auto tid = threadptr->tid();
        threadQue_.emplace(std::make_pair(tid, std::move(threadptr)));
    }
}

ThreadPool::~ThreadPool()
{
    cfg_.running_ = false;

    // Todo: How to design destruction
    cout << "ThreadPool::~ThreadPool()" << endl;   
    notEmptyCond_.notify_all();     // 获取任务
    notFullCond_.notify_all();      // 提交任务
}

Result ThreadPool::submit(std::shared_ptr<Task> task)
{
    if (!task)
    {
        #ifdef Pool_Exeception
            throw std::runtime_error("ThreadPool::submit() failed");
        #else
            return Result(task, ResultState::FAILURE);
        #endif
    }

    std::unique_lock<std::mutex> lock(taskMtx_);

    // 确保任务队列有空闲位置
    if (!notFullCond_.wait_for(lock, std::chrono::milliseconds(1000), 
        [this]()->bool 
        { 
            return cfg_.taskSize_ < cfg_.taskMax_ && cfg_.running_; 
        }))
    {
        // 用户线程被阻塞一秒, 但是还没有等到任务列队空余
        // 此时应该抛出异常或者打印错误, 但是返回空结果
        #ifdef Pool_Exeception
            throw std::runtime_error("ThreadPool::submit() failed");
        #else
            return Result(task, ResultState::FAILURE);
        #endif
    }
    
    taskQue_.emplace_back(task);
    cfg_.taskSize_ ++ ;

    if (cfg_.taskSize_ > 1)
        notFullCond_.notify_all();
    else
       notEmptyCond_.notify_one();

    return Result(task, ResultState::SUCCESS);
}

void ThreadPool::start()
{
    cfg_.running_ = true;

    for (const auto &[k, _] : threadQue_)
    {
        threadQue_[k]->start();
    }
}

void util::ThreadPool::consumerFunc(int tid)
{
    while (1)
    {
        // std::shared_ptr<Task> task = std::make_shared<Task>();
        std::shared_ptr<Task> task;
        {
            std::unique_lock<std::mutex> lock(taskMtx_);

            // 等待任务列队非空
            while (!cfg_.taskSize_)
            {
                if (!cfg_.running_)
                {
                    threadQue_.erase(tid); 
                    cfg_.currentThreadSize_ -- ;
                    return;
                }
                
                // 不能判断条件, 唤醒不仅仅由生产者唤醒, 还可能是因为销毁
                notEmptyCond_.wait(lock);
            }

            // 获取任务 
            task = taskQue_.front();
            taskQue_.pop_front();
            cfg_.taskSize_ -- ;
            
            if (!taskQue_.empty())
                notEmptyCond_.notify_all();

            notFullCond_.notify_all();
        }

        // exec
        if (task)
            task->execute();
    }
}
